package dyyx.zb;

import java.time.LocalDateTime;

import org.apache.flink.api.common.functions.GroupCombineFunction;
import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.core.fs.FileSystem.WriteMode;
import org.apache.flink.util.Collector;

public class EventsGroupAndSort2 {

	public static void main(String[] args) throws Exception {		
		System.out.println("EventsGroupAndSort2 start,"+LocalDateTime.now());
	
		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		String file = "file:/tmp/events5000w.txt";
		DataSet<Tuple5<String,String,String,String, Double>> csvInput = env.readCsvFile(file)
               .types(String.class,String.class,String.class,String.class, Double.class);
			
		csvInput.groupBy(2)
		.sortGroup(3, Order.ASCENDING)
		.sortGroup(1, Order.DESCENDING)
		
		
		// .first(9).print();
		// first 取每一个分组key的前N条记录
		
		.combineGroup(new MyGroupCombineFunction())
		
//		.print();
		.writeAsCsv("file:///tmp/events_result",WriteMode.OVERWRITE);
		env.execute();
		
		System.out.println("EventsGroupAndSort2 done,"+LocalDateTime.now());

	}
	
	private static class MyGroupCombineFunction implements GroupCombineFunction<Tuple5<String,String,String,String, Double>,Tuple5<String,String,String,String, Double>>{
		
		private static final long serialVersionUID = 1L;

		public void combine(Iterable<Tuple5<String,String,String,String, Double>> values, Collector<Tuple5<String,String,String,String, Double>> out) throws Exception{
			for(Tuple5<String,String,String,String, Double> item:values) {
				out.collect(item);
			}		
		}
	}
	
	

}
